package com.example.toy.core.index.impl;

import com.example.toy.core.dataset.BigCollection;
import com.example.toy.core.dataset.BigPairCollection;
import com.example.toy.core.dataset.Pair;
import com.example.toy.core.filter.ToyFilter;
import com.example.toy.core.fs.ToyTableVersionFileView;
import com.example.toy.core.index.ToyIndexDictionary;
import com.example.toy.core.reader.DefaultToyBaseFileReader;
import com.example.toy.core.reader.DefaultToyDeltaFileReader;
import com.example.toy.core.record.ToyRecord;
import com.example.toy.core.record.key.ToyRecordKey;
import com.example.toy.core.table.ToyTableMeta;
import com.example.toy.core.utils.RecordKeyHelper;
import com.example.toy.core.utils.StreamUtils;
import org.apache.hadoop.fs.Path;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * RealtimeToyIndexDictory
 *
 * @author shenbing
 * @date 2021/12/28
 */
public class NoneIndexDictionary implements ToyIndexDictionary {

  private final ToyTableMeta toyTableMeta;

  public NoneIndexDictionary(ToyTableMeta toyTableMeta) {
    this.toyTableMeta = toyTableMeta;
  }

  @Override
  public void addIndex(BigCollection<ToyRecordKey> key, Path path) {}

  @Override
  public BigPairCollection<ToyRecordKey, Iterator<Path>> findByIndex(
      BigCollection<ToyRecordKey> keys) {
    BigPairCollection<ToyRecordKey, Iterator<Path>> result =
        keys.groupBy(ToyRecordKey::getPartitionKey)
            .map(
                pair -> {
                  ToyTableVersionFileView fileView = this.toyTableMeta.getVersionFileView();
                  List<Path> baseFiles = fileView.getBaseFiles(pair.getLeft());

                  Map<ToyRecordKey, List<Path>> keyToPaths = new HashMap<>();
                  Optional<ToyFilter> reduce =
                      StreamUtils.toStream(pair.getRight())
                          .map(RecordKeyHelper::toToyFilter)
                          .reduce(ToyFilter::or);
                  for (Path baseFile : baseFiles) {
                    DefaultToyBaseFileReader reader =
                        new DefaultToyBaseFileReader(this.toyTableMeta, baseFile);
                    List<ToyRecord> foundRecords = reader.read(reduce.get());
                    for (ToyRecord foundRecord : foundRecords) {
                      List<Path> paths =
                          keyToPaths.computeIfAbsent(
                              foundRecord.getKey(), key -> new ArrayList<>());
                      paths.add(baseFile);
                    }
                  }
                  List<Path> deltaFiles = fileView.getDeltaFiles(pair.getLeft());
                  for (Path deltaFile : deltaFiles) {
                    DefaultToyDeltaFileReader reader =
                        new DefaultToyDeltaFileReader(this.toyTableMeta, deltaFile);
                    List<ToyRecord> foundRecords = reader.read(reduce.get());
                    for (ToyRecord foundRecord : foundRecords) {
                      List<Path> paths =
                          keyToPaths.computeIfAbsent(
                              foundRecord.getKey(), key -> new ArrayList<>());
                      paths.add(deltaFile);
                    }
                  }

                  return keyToPaths;
                })
            .flatMap(map -> map.entrySet().iterator())
            .mapToPair(it -> new Pair<>(it.getKey(), it.getValue().iterator()));

    return result;
  }

  @Override
  public void removeIndex(BigCollection<ToyRecordKey> key) {}

  @Override
  public void removeIndex(BigCollection<ToyRecordKey> key, Path path) {}

  @Override
  public void compact() {}
}
